/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.snda.lidejia.hivetotalsort.mr;

import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;

/**
 *
 * @author lidejia
 */
public class HiveSortKeyExtractor {

    private List<SortMeta> metas;

    public HiveSortKeyExtractor(JobConf conf) {
        String metaConfig = conf.get("hive.totalsort.meta");
        String[] metasStr = metaConfig.split(";");
        metas = new ArrayList<SortMeta>(metasStr.length);
        for (String metaStr : metasStr) {
            String[] meta = metaStr.split(":");
            metas.add(new SortMeta(Integer.parseInt(meta[0]), meta[1], meta[2]));
        }
    }

    public Writable[] extract(String line) {
        // split the columns
        String[] columns = line.split("\t");
        // result
        Writable[] writable = new Writable[metas.size()];
        for (int i = 0; i < writable.length; i++) {
            writable[i] = metas.get(i).parse(columns);
        }
        return writable;
    }
}
